package cn.itcast.flink.join;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * Author itcast
 * Date 2022/1/14 15:59
 * Desc 从socket获取数据，来计算传感器水位信息
 * 开发步骤：
 */
public class OutOfnessWindowDemo {
    public static void main(String[] args) throws Exception {
        //1.定义类 WaterSensor  String id; Long ts; Integer vc;
        //2.创建流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(1000);
        //3.获取socket文本数据
        DataStreamSource<String> source = env.socketTextStream("node1", 9999);
        //4.将字符串数据切分成 WaterSensor 对象数据
        //sensor_1,1547718199,35
        SingleOutputStreamOperator<WaterSensor> mapStream = source.map(new MapFunction<String, WaterSensor>() {
            @Override
            public WaterSensor map(String value) throws Exception {
                String[] arr = value.split(",");
                WaterSensor waterSensor = new WaterSensor(
                        arr[0],
                        Long.parseLong(arr[1]),
                        Integer.parseInt(arr[2])
                );
                return waterSensor;
            }
        });

        // 在窗口计算之前分配水印，防止数据出现乱序
        //5.分配水印机制，单调递增
        SingleOutputStreamOperator<WaterSensor> withWatermarkStream = mapStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((element, recordTimestamp) -> element.getTs() * 1000)
        );
        //6.分配后的数据根据id进行分组
        SingleOutputStreamOperator<String> result = withWatermarkStream.keyBy(t -> t.id)
                //7.设置滚动事件时间窗口，时间为10秒
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                //8.对开窗数据进行process
                .process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
                    @Override
                    public void process(String key, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
                        out.collect(
                                "key=" + key + "\n" +
                                "数据为:" + elements + "\n" +
                                "数量条数:" + elements.spliterator().estimateSize() + "\n" +
                                "窗口为:[" + context.window().getStart() + "," + context.window().getEnd() + ")\n" +
                                "=======================================================================\n\n");
                    }
                });
        //9.打印输出
        result.print();
        //10.执行流环境
        env.execute();
    }

    /**
     * 水位传感器：用于接收水位数据
     * id:传感器编号
     * ts:时间戳
     * vc:水位
     */
    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class WaterSensor {
        private String id;
        private Long ts;
        private Integer vc;
    }
}
